package cn.mesmile.flink.transform;

import cn.mesmile.flink.jdkstream.VideoOrder;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Date;

/**
 * @author zb
 * @date 2021/8/22 15:23
 * @Description map 适用于 一对一 转换
 */
public class FlinkTransformMapApp {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        env.setParallelism(2);

        DataStream<VideoOrder> ds = env.fromElements(
                new VideoOrder("21312","java",32,5,new Date()),
                new VideoOrder("314","java",32,5,new Date()),
                new VideoOrder("542","springboot",32,5,new Date()),
                new VideoOrder("42","redis",32,5,new Date()),
                new VideoOrder("52","java",32,5,new Date()),
                new VideoOrder("55","test",32,5,new Date()),
                new VideoOrder("56","cloud",32,5,new Date()),
                new VideoOrder("523","redis",32,5,new Date())
        );
        // map 适用于 一对一 转换
        DataStream<Tuple2<String, Integer>> mapDs = ds.map(new RichMapFunction<VideoOrder, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(VideoOrder videoOrder) throws Exception {
                return new Tuple2<>(videoOrder.getTitle(), 1);
            }
        });

        mapDs.print("result:");

        env.execute("custom map job");
    }
}
